/*
 * Copyright (C) 2010 Seapeak.Xu
 *
 * 5173mq client may be copied only under the terms of the GNU General
 * Public License V3, which may be found in the 5173mq source kit.
 *
 * Created on: 2010-10-27
 * Author: Seapeak.Xu
 */

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>

#include "fostmq_client.h"
#include "fost_sock.h"
#include "fost_netstream.h"
#include "fost_typedef.h"
#include "fost_msg.h"
#include "fostmq_protocol.h"
#include "fost_mem.h"
#include "fost_buff.h"
#include "fost_taskpool.h"
#include "fost_event.h"
#include "fost_logger.h"

int push_by_id(const char *ip,const int port,const int timeout,const char *id,const char *value)
{
	int result = 0; 
	int sockfd = client_sock_init(ip,port,timeout,true); 
	if(-1 == sockfd) 
	{ 
		printf("error:sockfd is %d.",sockfd); 
		return -1; 
	} 
	header_t header; 
	header.protocol = PUSHBYID; 
	header.state = 0; 
	int idlen = strlen(id); 
	int vallen = strlen(value); 
	header.bodylen = sizeof(int) + idlen +vallen; 
	char *buff = fost_callocs(HEADERLEN + header.bodylen);
	if(NULL == buff)
	{
		close(sockfd);
		return -1;
	}
	do
	{
		header2buff(&header,buff);
		int2buff(idlen,buff + HEADERLEN);
		memcpy(buff + HEADERLEN + sizeof(int),id,idlen);
		memcpy(buff + HEADERLEN + sizeof(int) + idlen,value,vallen);
		if(0 != send_by_select_nb(sockfd,timeout,buff,HEADERLEN + header.bodylen))
		{
			result = -1;
			break;
		}
		header_t resp_header;
		char header_buff[HEADERLEN + 1];
		memset(header_buff,0,HEADERLEN + 1);
		if(0 != recv_by_select_nb(sockfd,timeout,header_buff,HEADERLEN))
		{
			result = -1;
			break;//FOST_FREE(header_buff);
		}
		buff2header(&resp_header,header_buff);
		result = resp_header.state;
	}while(false);
	close(sockfd);
	FOST_FREE(buff);
	return result;
}

int push_by_name(const char *ip,const int port,const int timeout,const char *name,const char *value) 
{ 
	int result = 0; 
	int sockfd = client_sock_init(ip,port,timeout,true); 
	if(-1 == sockfd) 
	{ 
		printf("error:sockfd is %d.",sockfd); 
		return -1; 
	} 
	header_t header; 
	header.protocol = PUSHBYNAME; 
	header.state = 0; 
	int namelen = strlen(name); 
	int vallen = strlen(value); 
	header.bodylen = sizeof(int) + namelen +vallen; 
	char *buff = fost_callocs(HEADERLEN + header.bodylen);
	if(NULL == buff)
	{
		close(sockfd);
		return -1;
	}
	do
	{
		header2buff(&header,buff);
		int2buff(namelen,buff + HEADERLEN);
		memcpy(buff + HEADERLEN + sizeof(int),name,namelen);
		memcpy(buff + HEADERLEN + sizeof(int) + namelen,value,vallen);
		if(0 != send_by_select_nb(sockfd,timeout,buff,HEADERLEN + header.bodylen))
		{
			result = -1;
			break;
		}
		header_t resp_header;
		char header_buff[HEADERLEN + 1];
		memset(header_buff,0,HEADERLEN + 1);
		if(0 != recv_by_select_nb(sockfd,timeout,header_buff,HEADERLEN))
		{
			result = -1;
			break;//FOST_FREE(header_buff);
		}
		buff2header(&resp_header,header_buff);
		result = resp_header.state;
	}while(false);
	close(sockfd);
	FOST_FREE(buff);
	return result;

} 

int pop_by_id(const char *ip,const int port,const int timeout,const char *id,char *value) 
{ 
	int result = 0; 
	int sockfd = client_sock_init(ip,port,timeout,true); 
	if(-1 == sockfd) 
	{ 
		printf("error:sockfd is %d.",sockfd);
		return -1; 
	} 
	header_t header; 
	header.protocol = POPBYID; 
	header.state = 0; 
	int idlen = strlen(id); 
	header.bodylen = idlen; 
	char *buff = fost_callocs(HEADERLEN + header.bodylen);
	if(NULL == buff)
	{
		close(sockfd);
		return -1;
	}
	do
	{
		header2buff(&header,buff);
		memcpy(buff + HEADERLEN ,id,idlen);
		if(0 != send_by_select_nb(sockfd,timeout,buff,HEADERLEN + header.bodylen))
		{
			result = -1;
			break;
		}
		header_t resp_header;
		char header_buff[HEADERLEN + 1];
		memset(header_buff,0,HEADERLEN + 1);
		if(0 != recv_by_select_nb(sockfd,timeout,header_buff,HEADERLEN))
		{
			result = -1;
			break;
		}
		buff2header(&resp_header,header_buff);
		if(0 != resp_header.state)
		{
			result = -1;
			break;
		}

		if(0 != recv_by_select_nb(sockfd,timeout,value,resp_header.bodylen))
		{
			result = -1;
			break;
		}
	}while(false);
	close(sockfd);
	FOST_FREE(buff);
	return result;

}

int pop_and_delete_by_id(const char *ip,const int port,const int timeout,const char *id,char *value) 
{ 
	int result = 0; 
	int sockfd = client_sock_init(ip,port,timeout,true); 
	if(-1 == sockfd) 
	{ 
		printf("error:sockfd is %d.",sockfd);
		return -1; 
	} 
	header_t header; 
	header.protocol = POPDELBYID; 
	header.state = 0; 
	int idlen = strlen(id); 
	header.bodylen = idlen; 
	char *buff = fost_callocs(HEADERLEN + header.bodylen);
	if(NULL == buff)
	{
		close(sockfd);
		return -1;
	}
	do
	{
		header2buff(&header,buff);
		memcpy(buff + HEADERLEN ,id,idlen);
		if(0 != send_by_select_nb(sockfd,timeout,buff,HEADERLEN + header.bodylen))
		{
			result = -1;
			break;
		}
		header_t resp_header;
		char header_buff[HEADERLEN + 1];
		memset(header_buff,0,HEADERLEN + 1);
		if(0 != recv_by_select_nb(sockfd,timeout,header_buff,HEADERLEN))
		{
			result = -1;
			break;
		}
		buff2header(&resp_header,header_buff);
		if(0 != resp_header.state)
		{
			result = -1;
			break;
		}

		if(0 != recv_by_select_nb(sockfd,timeout,value,resp_header.bodylen))
		{
			result = -1;
			break;
		}
	}while(false);
	close(sockfd);
	FOST_FREE(buff);
	return result;

}



int pop_by_name(const char *ip,const int port,const int timeout,const char *name,char *value) 
{ 
	int result = 0; 
	int sockfd = client_sock_init(ip,port,timeout,true); 
	if(-1 == sockfd) 
	{ 
		printf("error:sockfd is %d.",sockfd);
		return -1; 
	} 
	header_t header; 
	header.protocol = POPBYNAME; 
	header.state = 0; 
	int namelen = strlen(name); 
	header.bodylen = namelen; 
	char *buff = fost_callocs(HEADERLEN + header.bodylen);
	if(NULL == buff)
	{
		close(sockfd);
		return -1;
	}
	do
	{
		header2buff(&header,buff);
		memcpy(buff + HEADERLEN ,name,namelen);
		if(0 != send_by_select_nb(sockfd,timeout,buff,HEADERLEN + header.bodylen))
		{
			result = -1;
			break;
		}
		header_t resp_header;
		char header_buff[HEADERLEN + 1];
		memset(header_buff,0,HEADERLEN + 1);
		if(0 != recv_by_select_nb(sockfd,timeout,header_buff,HEADERLEN))
		{
			result = -1;
			break;
		}
		buff2header(&resp_header,header_buff);
		if(0 != resp_header.state)
		{
			result = -1;
			break;
		}

		if(0 != recv_by_select_nb(sockfd,timeout,value,resp_header.bodylen))
		{
			result = -1;
			break;
		}
	}while(false);
	close(sockfd);
	FOST_FREE(buff);
	return result;

}

int delete_by_id(const char *ip,const int port,const int timeout,const char *id)
{
	int result = 0; 
	int sockfd = client_sock_init(ip,port,timeout,true); 
	if(-1 == sockfd) 
	{ 
		printf("error:sockfd is %d.",sockfd);
		return -1; 
	} 
	header_t header; 
	header.protocol = DELETEBYID; 
	header.state = 0; 
	int idlen = strlen(id); 
	header.bodylen = idlen; 
	char *buff = fost_callocs(HEADERLEN + header.bodylen);
	if(NULL == buff)
	{
		close(sockfd);
		return -1;
	}
	do
	{
		header2buff(&header,buff);
		memcpy(buff + HEADERLEN ,id,idlen);
		if(0 != send_by_select_nb(sockfd,timeout,buff,HEADERLEN + header.bodylen))
		{
			result = -1;
			break;
		}
		header_t resp_header;
		char header_buff[HEADERLEN + 1];
		memset(header_buff,0,HEADERLEN + 1);
		if(0 != recv_by_select_nb(sockfd,timeout,header_buff,HEADERLEN))
		{
			result = -1;
			break;
		}
		buff2header(&resp_header,header_buff);
		if(0 != resp_header.state)
		{
			result = -1;
			break;
		}
	}while(false);
	close(sockfd);
	FOST_FREE(buff);
	return result;
}

int destroy_queue(const char *ip,const int port,const int timeout,const char *name)
{
	int result = 0; 
	int sockfd = client_sock_init(ip,port,timeout,true); 
	if(-1 == sockfd) 
	{ 
		printf("error:sockfd is %d.",sockfd);
		return -1; 
	} 
	header_t header; 
	header.protocol = DESTROYQUEUE; 
	header.state = 0; 
	int namelen = strlen(name); 
	header.bodylen = namelen; 
	char *buff = fost_callocs(HEADERLEN + header.bodylen);
	if(NULL == buff)
	{
		close(sockfd);
		return -1;
	}
	do
	{
		header2buff(&header,buff);
		memcpy(buff + HEADERLEN ,name,namelen);
		if(0 != send_by_select_nb(sockfd,timeout,buff,HEADERLEN + header.bodylen))
		{
			result = -1;
			break;
		}
		header_t resp_header;
		char header_buff[HEADERLEN + 1];
		memset(header_buff,0,HEADERLEN + 1);
		if(0 != recv_by_select_nb(sockfd,timeout,header_buff,HEADERLEN))
		{
			result = -1;
			break;
		}
		buff2header(&resp_header,header_buff);
		if(0 != resp_header.state)
		{
			result = -1;
			break;
		}
	}while(false);
	close(sockfd);
	FOST_FREE(buff);
	return result;
}


